/*
 *  Queen Mary University of London MOM System
 *  
 *  Copyright Queen Mary University of London
 *  
 *  Authors:
 *  Bob Chew (bob.chew@elec.qmul.ac.uk)
 *  Beatriz Viñal Murciano (beavimu@gmail.com)
 *  John Bigham (john.bigham@elec.qmul.ac.uk)
 *  
 *  
 *  Portions of this software uses Apache Qpid
 *    http://qpid.apache.org
 *  
 *  Apache Qpid license:
 * --------------------------------------------------------------
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 * -------------------------------------------------------------
 * 
 */
package brokerBottleneckDetector;

import java.lang.management.*;
import javax.management.MBeanServerConnection;
import com.sun.management.OperatingSystemMXBean;

import java.text.DecimalFormat;
import java.text.SimpleDateFormat;
import java.util.UUID;

/**
 * Sends AMQP messages to itself through a broker and gets an idea of how well
 * that broker is working based on whether it gets the message or not. Calls the
 * updateState method in the calling Detector.
 */
public class DetectorThread extends Thread {
	private boolean __DEBUG = false;
	
	/** Detector. */
	private Detector bbDetector;

	/** Unique ID of the client. */
	private final String clientID;
	/** Name of the used topic. */
	private final String topicName;
	 /** Machine in which the broker to test is running. */
	 private final String broker;
	 /** Port in which the broker to test is running. */
	 private final int port;

	private final MBeanServerConnection mJVM;
	private static final String jmxHost = "-host ";
	private static final String jmxPort = "-port ";
	private static final String jmxUser = "admin"; // remember to set this in
	// Qpid conf first!
	private static final String jmxPass = "admin"; // remember to set this in
	// Qpid conf first!

	/** Time in which a message must arrive, in milliseconds. */
	// private static final int MAX_WAIT = 10000;
	/** Time between messages. */
	private static final int INTERVAL = 1000;
	/** Number of records to take before sending to Detector */
	private static final int NUM_RECORDS = 10;

	/**
	 * Constructor.
	 * 
	 * @param bfDetector
	 *            to report the state of the broker to
	 * @param testBroker
	 *            IP address of the broker to test
	 * @param port
	 *            port in which the broker to test is listening
	 * @param monitoredJVM
	 *            a MBeanServerConnection to the broker
	 */
	public DetectorThread(final Detector bbDetector, final String testBroker,
			final int testPort) {

		this.bbDetector = bbDetector;
		clientID = UUID.randomUUID().toString();
		topicName = "brokerBottleneck." + clientID;
		this.broker = testBroker;
		this.port = testPort;

		JMXhelper j = new JMXhelper();
		this.mJVM = j.getJVMBeanSecure(jmxHost + broker + " " + jmxPort
				+ port, jmxUser, jmxPass);

		start();

	}

	// Variables for calculating performance figures
	// Ugly, but meh!
	private static class CPUTime {
		long upTime = -1L;
		long processCpuTime = -1L;
		long timeStamp;
		int nCPUs;
	}


	boolean ewmaRun = false; // wait for the initial window to be over
	int ewmaWindow = 1;
	double ewmaAlpha = 0.6;
	double ewmaPrevResidue = 0d;
	double currEwma = 0;

	String records = "";
	int recNum = 0;
	
	long prevUpTime, prevProcessCpuTime = 0;

	private synchronized void perfstats() {
		BrokerMXBeans bmxb = new BrokerMXBeans();
		boolean send = false;

		// Grab the various MXBeans and stuff them into BrokerMXBeans
		// Wanted MXBeans:
		// - Runtime
		// - Operating System
		// - Memory
		try {
			RuntimeMXBean rtb = ManagementFactory.newPlatformMXBeanProxy(mJVM,
					ManagementFactory.RUNTIME_MXBEAN_NAME, RuntimeMXBean.class);

			OperatingSystemMXBean osb = ManagementFactory
					.newPlatformMXBeanProxy(mJVM,
							ManagementFactory.OPERATING_SYSTEM_MXBEAN_NAME,
							OperatingSystemMXBean.class);

			MemoryMXBean memb = ManagementFactory.newPlatformMXBeanProxy(mJVM,
					ManagementFactory.MEMORY_MXBEAN_NAME, MemoryMXBean.class);

			bmxb.memorymxbean = memb;
			bmxb.operatingsystemmxbean = osb;
			bmxb.runtimemxbean = rtb;

		} catch (Exception e) {
			e.printStackTrace();
		}

		CPUTime cputime = new CPUTime();

		// get uptime from RuntimeMXBean
		// uptime refers to how long the process has been running
		// use this figure in the calculations for instantaeneous CPU usage
		cputime.upTime = bmxb.runtimemxbean.getUptime();
		cputime.processCpuTime = bmxb.operatingsystemmxbean.getProcessCpuTime();
		cputime.timeStamp = System.currentTimeMillis();
//		cputime.nCPUs = bmxb.operatingsystemmxbean.getAvailableProcessors();  // hyperthreading will confuse this...
		cputime.nCPUs = 1;

		// only activate if we are not in the past
		if (prevUpTime > 0L && cputime.upTime > prevUpTime) {
			// elapsedCpu is in ns and elapsedTime is in ms.
			long elapsedCpu = cputime.processCpuTime - prevProcessCpuTime;
			long elapsedTime = cputime.upTime - prevUpTime;

			// cpuUsage could go higher than 100% because elapsedTime
			// and elapsedCpu are not fetched simultaneously
			float cpuUsage = Math.min(100F, elapsedCpu
					/ (elapsedTime * 10000F * cputime.nCPUs));
			
			// EWMA
			if (ewmaRun) {
				currEwma = (ewmaAlpha * cpuUsage)
						+ ((1 - ewmaAlpha) * ewmaPrevResidue);
			} else {
				ewmaPrevResidue = cpuUsage;
				ewmaRun = true;
			}

			// Memory statistics
//			MemoryUsage mHeap = bmxb.memorymxbean.getHeapMemoryUsage();
//			MemoryUsage mNonHeap = bmxb.memorymxbean.getNonHeapMemoryUsage();

			// printing...
			SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd HH:mm:ss.SS");
			DecimalFormat df = new DecimalFormat("##0.00");
			// tabby tabby... human readable
//			if(__DEBUG)
//			System.out.println(sdf.format(cputime.timeStamp) + "\t"
//					+ df.format(cpuUsage) + "\t" + df.format(currEwma) + "\t\t"
//					+ mHeap.getUsed() + "\t" + mHeap.getCommitted() + "\t"
//					+ mHeap.getMax() + "\t" + mNonHeap.getUsed() + "\t"
//					+ mNonHeap.getCommitted() + "\t" + mNonHeap.getMax());
			// CSV string. Bayes Classifier has methods to handle CSV inputs
			String out = sdf.format(cputime.timeStamp) + ","
					+ df.format(cpuUsage) + "," + df.format(currEwma) + ","
					+ elapsedCpu + "," + cputime.processCpuTime + "," + prevProcessCpuTime + "\n";
//					+ mHeap.getUsed() + "," + mHeap.getCommitted() + ","
//					+ mHeap.getMax() + "," + mNonHeap.getUsed() + ","
//					+ mNonHeap.getCommitted() + "," + mNonHeap.getMax() + "\n";

			records += out;
			recNum++;
			if (recNum >= NUM_RECORDS) {
				send = true;
			}

//			if(__DEBUG) System.out.print(out); // debug
		}
		prevUpTime = cputime.upTime;
		prevProcessCpuTime = cputime.processCpuTime;

		// call the Detector to do make a classification
		if (send) {
			try {
				// call the decision making method in Detector
				bbDetector.testBottleneck(broker + ":" + port, records);
			} catch (Exception e) {
				e.printStackTrace();
			}
			
			send = false;
			records = "";
			recNum = 0;
//			yield();
		}
	}

	/**
	 */
	@Override
	public void run() {
		try {
			while (true) {
				perfstats();
				sleep(INTERVAL);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}
